package com.zch.rpc.register.register;

import java.io.IOException;
import java.net.Socket;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 配置服务注册
 *
 */
public class ServiceRegistry {
	private String host;
	private Integer port;

	private Executor threadPoolService;
	private ZkClient zkClient;
	public ServiceRegistry(String host, Integer port) {
		super();
		this.host = host;
		this.port = port;
		threadPoolService= new ThreadPoolExecutor(12, 15, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
	}

	public ZkClient getZkClient() {
		return zkClient;
	}

	public void setZkClient(ZkClient zkClient) {
		this.zkClient = zkClient;
	}

	/**
	 * 注册服务
	 * @param serviceName 服务名称
	 * @param serverAddress RPC服务地址
	 * @throws IOException 
	 */
	public void register(String serviceName, String serverAddress) throws IOException {
		//一个接口一个socket通道
		ZkClient  zkClient = new ZkClient(this.host, this.port);//2181
		Message msg = new Message();
		String[] array = serviceName.split(":");

		if (array.length == 2) {
			msg.setInterfaceName(array[0]);
			msg.setVersion(array[1]);			
		} else {
			msg.setInterfaceName(array[0]);
			msg.setVersion(null);	
		}
		msg.setServiceAddress(serverAddress);
		msg.setType("register");
		zkClient.register(msg);
		//定期发送心跳
		threadPoolService.execute(new Runnable() {
			@Override
			public void run() {
				while(true){
					try {
						Thread.sleep(2000);

						zkClient.keepAlive();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		});
		//重连
		threadPoolService.execute(new Runnable() {
			@Override
			public void run() {
				while (true){
					if(zkClient.reconnect){
						if(zkClient.reconnectCount<=3){
							try {
								register(serviceName,serverAddress);
								zkClient.reconnectCount=0;
								zkClient.reconnect=false;
								System.out.println("客户端重连成功");
							}catch (Exception e){
								e.printStackTrace();
								zkClient.reconnectCount++;
							}
						}else{
							System.out.println("已重连三次，请检查zookeeper");
						}
					}
				}
			}
		});

	}
	public boolean keepAlive(){
		return zkClient.keepAlive();
	}

    /*public void reconnect(String serviceName,int port) throws IOException {
		Set<String> keyset=services.keySet();
		for(String s:keyset){
			register(s,"localhost:"+port);
		}
	}*/
	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public Integer getPort() {
		return port;
	}

	public void setPort(Integer port) {
		this.port = port;
	}
	
	

}
